package com.cc.flink;

import com.cc.flink.config.JobConfig;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.cc.flink.sink.DorisBulkJdbcSink;

/**
 * Flink作业主类：实现Kafka到Doris的数据批处理管道
 * <p>
 * 主要功能：
 * 1. 从Kafka实时读取数据
 * 2. 进行数据过滤和格式转换
 * 3. 使用自定义窗口和触发器进行批处理
 * 4. 批量写入Doris数据库
 * 5. 支持检查点机制，实现精确一次处理语义
 * <p>
 * 设计模式：采用函数式编程模式，通过组合各种Flink操作符构建数据处理管道
 */
public class KafkaToDorisJob {

    /**
     * 日志记录器，用于记录作业运行状态和错误信息
     */
    private static final Logger logger = LoggerFactory.getLogger(KafkaToDorisJob.class);

    /**
     * 程序入口方法，构建并执行Flink数据处理管道
     * @param args 命令行参数（当前未使用）
     * @throws Exception 当作业执行出错时抛出异常
     */
    public static void main(String[] args) throws Exception {
        // 初始化配置，从环境变量加载配置参数或使用默认值
        JobConfig config = new JobConfig();
        
        // 创建Flink流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 注意：在Flink 1.12+版本中，默认使用处理时间语义，不再需要显式设置TimeCharacteristic
        
        // 配置检查点机制，确保作业的容错性和状态一致性
        configureCheckpoint(env, config);
        
        // 创建Kafka数据源，配置从Kafka读取数据的参数
        KafkaSource<String> kafkaSource = createKafkaSource(config);
        // 将KafkaSource转换为Flink的DataStream
        DataStream<String> sourceStream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source");
        
        // 构建数据处理管道
        // 数据处理流程：过滤 -> 转换格式 -> 窗口聚合 -> 批量写入Doris
        DataStream<Tuple2<String, String>> dataStream = sourceStream
            // 第一步：数据过滤，移除无效数据
            .filter(new DataFilter())
            // 第二步：数据转换，将字符串转换为Tuple2格式
            .map(new DataTransformer())
            // 第三步：按业务键分区，确保相关数据进入同一处理窗口
            .keyBy(value -> value.f0)
            // 第四步：使用滚动处理时间窗口
            .window(TumblingProcessingTimeWindows.of(Time.seconds(config.getWindowSizeSeconds())))
            // 第五步：应用自定义触发器，支持基于记录数或时间的触发
            .trigger(new CountOrTimeTrigger(config.getMaxRecordsPerBatch(), config.getWindowSizeSeconds() * 1000))
            // 第六步：窗口内数据处理
            .process(new BatchProcessFunction());
        
        // 最后一步：将处理后的数据批量写入Doris数据库
        createDorisJdbcSink(config, dataStream);
        
        // 执行Flink作业，启动数据处理流程
        env.execute("Kafka to Doris Batch Job");
    }
    
    /**
     * 配置Flink检查点参数，是保证作业容错性的关键
     * @param env 流执行环境
     * @param config 作业配置
     * @throws Exception 配置过程中的异常
     */
    private static void configureCheckpoint(StreamExecutionEnvironment env, JobConfig config) throws Exception {
        // 启用检查点机制，设置检查点间隔（毫秒）
        env.enableCheckpointing(config.getCheckpointIntervalMs());
        
        // 获取检查点配置对象
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        
        // 设置检查点模式为EXACTLY_ONCE，保证数据仅被处理一次
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        
        // 设置两次检查点之间的最小间隔，防止检查点过于密集影响性能
        checkpointConfig.setMinPauseBetweenCheckpoints(5000);
        
        // 设置检查点超时时间，超过此时间未完成的检查点将被取消
        checkpointConfig.setCheckpointTimeout(config.getCheckpointTimeoutMs());
        
        // 设置允许同时进行的最大检查点数
        checkpointConfig.setMaxConcurrentCheckpoints(config.getMaxConcurrentCheckpoints());
        
        // 配置作业取消时检查点的行为：保留外部检查点，以便从检查点恢复
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        
        // 配置状态后端为HashMapStateBackend，适用于中小规模状态，性能较好
        env.setStateBackend(new HashMapStateBackend());
        // 配置检查点存储位置
        env.getCheckpointConfig().setCheckpointStorage(config.getCheckpointPath());
    }
    
    /**
     * 创建Kafka数据源，配置Kafka连接参数和消费行为
     * @param config 作业配置
     * @return 配置好的KafkaSource实例
     */
    private static KafkaSource<String> createKafkaSource(JobConfig config) {
        // 使用KafkaSource构建器模式创建数据源
        return KafkaSource.<String>
            builder()
            // 设置Kafka集群地址
            .setBootstrapServers(config.getKafkaBootstrapServers())
            // 设置要消费的Kafka主题
            .setTopics(config.getKafkaTopic())
            // 设置消费者组ID
            .setGroupId(config.getKafkaGroupId())
            // 设置消费起始偏移量：从已提交的偏移量开始消费，支持从保存点恢复
            .setStartingOffsets(OffsetsInitializer.committedOffsets())
            // 设置值反序列化器，将Kafka消息转换为字符串
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .build();
    }
    
    /**
     * 创建Doris JDBC Sink，配置数据写入行为
     * @param config 作业配置
     * @param processedStream 处理后的数据流
     */
    private static void createDorisJdbcSink(JobConfig config, DataStream<Tuple2<String, String>> processedStream) {
        // 创建自定义的Doris批量写入Sink
        // 配置：作业配置、批处理大小、刷新间隔、最大重试次数
        DorisBulkJdbcSink dorisSink = new DorisBulkJdbcSink(
                config,
                (int) config.getMaxRecordsPerBatch(),
                config.getBatchWindowTimeMs(),
                3 // 最大重试次数，提高写入可靠性
        );
        
        // 将Sink添加到处理管道中
        processedStream.addSink(dorisSink)
                .name("Doris-Bulk-JDBC-Sink") // 为操作符添加名称，便于监控和调试
                .uid("doris-bulk-sink-id") // 设置唯一ID，保证状态一致性和可恢复性
                .setParallelism(1); // 设置并行度，根据Doris集群规模调整
    }
    
    /**
     * 数据过滤函数
     * 实现数据清洗，过滤掉无效或空的记录，避免下游处理异常数据
     */
    public static class DataFilter implements FilterFunction<String> {
        /**
         * 过滤逻辑实现
         * @param value 输入的字符串数据
         * @return true表示数据有效，false表示数据无效
         * @throws Exception 处理过程中的异常
         */
        @Override
        public boolean filter(String value) throws Exception {
            // 检查数据是否为null，并且移除前后空格后不为空字符串
            return value != null && !value.trim().isEmpty();
        }
    }
    
    /**
     * 数据转换函数
     * 将输入的字符串数据转换为结构化的Tuple2格式，便于后续处理
     */
    public static class DataTransformer implements MapFunction<String, Tuple2<String, String>> {
        /**
         * 转换逻辑实现
         * @param value 输入的字符串数据
         * @return 转换后的Tuple2对象，包含id和data两部分
         */
        @Override
        public Tuple2<String, String> map(String value) {
            // 示例实现：按逗号分割数据，分为id和data两部分
            // 实际业务场景中应根据具体数据格式进行适当的解析和转换
            String[] parts = value.split(",", 2);  // 最多分割成两部分，避免多个逗号导致的数据丢失
            String id = parts.length > 0 ? parts[0] : "";  // 第一部分作为业务主键id
            String data = parts.length > 1 ? parts[1] : value;  // 第二部分或整个字符串作为数据内容
            return new Tuple2<>(id, data);  // 返回结构化的数据对
        }
    }
    
    /**
     * 批量处理函数
     * 在窗口触发时处理窗口内的所有数据，是数据聚合和转换的关键环节
     */
    public static class BatchProcessFunction extends ProcessWindowFunction<Tuple2<String, String>, Tuple2<String, String>, String, TimeWindow> {
        /**
         * 窗口处理逻辑实现
         * @param key 窗口的分组键，这里是Tuple2的第一个元素(id)
         * @param context 窗口上下文，包含窗口信息和时间戳等
         * @param elements 窗口内的所有元素
         * @param out 结果收集器，用于输出处理后的元素
         */
        @Override
        public void process(String key, Context context, Iterable<Tuple2<String, String>> elements, Collector<Tuple2<String, String>> out) {
            // 当前实现：简单转发每个元素，不做额外处理
            // 实际应用中可以在此处进行复杂的聚合运算、业务逻辑处理等
            for (Tuple2<String, String> element : elements) {
                out.collect(element);
            }
            
            // 记录窗口处理日志，包含键和窗口时间范围信息
            logger.info("Processed window with key {}, window start: {}, window end: {}", 
                key, context.window().getStart(), context.window().getEnd());
        }
    }
    
    /**
     * 自定义触发器
     * 实现基于记录数量和时间双重触发机制，支持尽快处理数据的同时控制批大小
     */
    public static class CountOrTimeTrigger extends org.apache.flink.streaming.api.windowing.triggers.Trigger<Tuple2<String, String>, TimeWindow> {
        private final long maxCount;  // 触发窗口的最大元素数量
        private final long maxTime;   // 触发窗口的最大时间间隔（毫秒）
        
        /**
         * 状态描述符定义，用于存储当前窗口的元素计数状态
         * 该状态会参与检查点机制，保证故障恢复时计数的准确性
         */
        private final ListStateDescriptor<Long> countStateDesc = 
            new ListStateDescriptor<>("count", org.apache.flink.api.common.typeinfo.Types.LONG);
        
        /**
         * 构造函数
         * @param maxCount 最大元素数量阈值
         * @param maxTime 最大时间阈值（毫秒）
         */
        public CountOrTimeTrigger(long maxCount, long maxTime) {
            // 参数校验，确保配置有效
            Preconditions.checkArgument(maxCount > 0, "maxCount must be greater than 0");
            Preconditions.checkArgument(maxTime > 0, "maxTime must be greater than 0");
            this.maxCount = maxCount;
            this.maxTime = maxTime;
        }
        
        /**
         * 当新元素到达窗口时调用
         * 负责更新计数并决定是否触发窗口
         */
        @Override
        public TriggerResult onElement(Tuple2<String, String> element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
            // 获取当前分区的状态
            ListState<Long> countState = ctx.getPartitionedState(countStateDesc);
            
            // 读取并更新当前计数
            Long count = 0L;
            if (countState.get().iterator().hasNext()) {
                count = countState.get().iterator().next();
            }
            count += 1;
            // 更新状态存储
            countState.clear();
            countState.add(count);
            
            // 注册处理时间定时器，确保即使数据量不足也能定时处理
            ctx.registerProcessingTimeTimer(window.maxTimestamp());
            
            // 当元素数量达到阈值时触发窗口计算并清除状态
            if (count >= maxCount) {
                logger.info("Triggering window due to count threshold: {}", count);
                countState.clear();  // 重置计数，准备下一个窗口
                return TriggerResult.FIRE_AND_PURGE;  // FIRE_AND_PURGE表示触发计算并清除窗口内容
            }
            // 否则继续收集元素
            return TriggerResult.CONTINUE;
        }
        
        /**
         * 当处理时间定时器触发时调用
         * 用于实现基于时间的窗口触发
         */
        @Override
        public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            // 记录时间触发日志
            logger.info("Triggering window due to time threshold");
            // 清除计数状态
            ListState<Long> countState = ctx.getPartitionedState(countStateDesc);
            countState.clear();
            // 触发窗口计算并清除窗口内容
            return TriggerResult.FIRE_AND_PURGE;
        }
        
        /**
         * 当事件时间定时器触发时调用
         * 由于使用处理时间窗口，此方法返回CONTINUE
         */
        @Override
        public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.CONTINUE;  // 不基于事件时间触发
        }
        
        /**
         * 清理窗口资源
         * 当窗口被移除时调用，确保释放所有相关资源
         */
        @Override
        public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
            // 删除注册的定时器，避免内存泄漏
            ctx.deleteProcessingTimeTimer(window.maxTimestamp());
            // 清除状态，释放资源
            ListState<Long> countState = ctx.getPartitionedState(countStateDesc);
            countState.clear();
        }
    }
}